package com.bw.gmall.realtime.dwd.db.split.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.common.base.BaseApp;
import com.bw.gmall.realtime.common.bean.TableProcessDwd;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.FlinkSinkUtil;
import com.bw.gmall.realtime.common.util.FlinkSourceUtil;
import com.bw.gmall.realtime.dwd.db.split.function.BaseDbTableProcessFunction;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;


public class DwdBaseDb extends BaseApp {
    public static void main(String[] args) {
        new DwdBaseDb().start(10019,4, "dwd_base_db",Constant.TOPIC_DB);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //TODO 对数据中的流数据镜像类型转换并进行简单的ETL jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = etl(stream);

//        jsonObjDS.print();
        //TODO 使用Flink读取配置表中的信息
        //创建MysqlSource对象
        MySqlSource<String> mysqlSource = FlinkSourceUtil.getMysqlSource(Constant.PROCESS_DATABASE, Constant.PROCESS_DWD_TABLE_NAME);
        //读取数据 封装为流
        DataStreamSource<String> mysqlStrDS = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "mysql_source");
        //对流中的数据进行类型转换  jsonStr -> 实体类对象
        SingleOutputStreamOperator<TableProcessDwd> tbDS = mysqlStrDS.map(new MapFunction<String, TableProcessDwd>() {
            @Override
            public TableProcessDwd map(String value) throws Exception {
                // 为了处理方便,先将jsonStr封装为JsonObj
                JSONObject jsonObj = JSON.parseObject(value);
                //获取操作类型
                String op = jsonObj.getString("op");
                TableProcessDwd tp = null;
                if ("d".equals(op)){
                    // 对配置表进行了删除操作   需要从before属性中删除配置信息
                    tp = jsonObj.getObject("before",TableProcessDwd.class);
                }else {
                    // 对配置表进行了读,插入,更新操作  需要从 after属性中获取配置信息
                    tp = jsonObj.getObject("after",TableProcessDwd.class);
                }
                tp.setOp(op);
                return tp;
            }
        });
        //TODO 对配置流进行广播  --broadcast
        MapStateDescriptor<String, TableProcessDwd> mapStateDescriptor = new MapStateDescriptor<String, TableProcessDwd>(
                "mapStateDescriptor",String.class,TableProcessDwd.class
        );
        BroadcastStream<TableProcessDwd> broadcastDS = tbDS.broadcast(mapStateDescriptor);
        //todo 关联主流业务数据和广播流中的配置数据 --connect
        BroadcastConnectedStream<JSONObject, TableProcessDwd> connectDS = jsonObjDS.connect(broadcastDS);
        //todo 对关联后的数据进行处理  --process
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcessDwd>> splitDS = connectDS.process(new BaseDbTableProcessFunction(mapStateDescriptor));
        //todo 将处理逻辑比较简单的事实表数据写到kafka的不同主题中
//        splitDS.print();
        splitDS.sinkTo(FlinkSinkUtil.getKafkaSink());




    }

    private static SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(jsonStr);
                    String type = jsonObject.getString("type");
                    if (!type.startsWith("bootstrap-")) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    throw new RuntimeException("这不是一个标准的json");
                }
            }
        });
    }

}
